package com.insight.common.mq;

import io.lettuce.core.Consumer;
import io.lettuce.core.Limit;
import io.lettuce.core.Range;
import io.lettuce.core.api.async.RedisAsyncCommands;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

@Slf4j
public class PendingAckCleaner {

    private final RedisAsyncCommands<String, String> async;
    private final String stream;
    private final String group;
    private final long idleTimeoutMs;
    private final int pageSize;
    private final ExecutorService executor;
    private final ScheduledExecutorService scheduler;
    private final String cleanerConsumer;

    public PendingAckCleaner(RedisAsyncCommands<String, String> async,
                             String stream,
                             String group,
                             String cleanerConsumer,
                             long idleTimeoutMs,
                             long checkIntervalMs,
                             int pageSize,
                             int parallelism) {
        this.async = async;
        this.stream = stream;
        this.group = group;
        this.cleanerConsumer = cleanerConsumer;
        this.idleTimeoutMs = idleTimeoutMs;
        this.pageSize = pageSize;
        this.executor = Executors.newFixedThreadPool(parallelism);
        this.scheduler = Executors.newSingleThreadScheduledExecutor();

        scheduler.scheduleAtFixedRate(this::scanAndAckAll, 0, checkIntervalMs, TimeUnit.MILLISECONDS);
    }

    private void scanAndAckAll() {
        executor.submit(() -> fetchAndAckBatch("-", "+"));
    }

    private void fetchAndAckBatch(String startId, String endId) {
        Consumer<String> consumer = Consumer.from(group, cleanerConsumer);
        async.xpending(stream, consumer, Range.create(startId, endId), Limit.from(pageSize))
                .thenAccept(list -> {
                    if(list.isEmpty()) return;
                    List<String> expiredIds = getExpiredIds(list);

                    if (!expiredIds.isEmpty()) {
                        async.xack(stream, group, expiredIds.toArray(new String[0]));
                        log.info("[{}/{}] 已 ACK 超时消息 {} 条", stream, group, expiredIds.size());
                    }
                    // 如果满页，拉下一页
                    if (list.size() == pageSize) {
                        String lastId = lastId(list);
                        if (StringUtils.isNotBlank(lastId)) {
                            String nextStartId = incrementId(lastId);
                            executor.submit(() -> fetchAndAckBatch(nextStartId, endId));
                        }
                    }
                })
                .exceptionally(ex -> {
                    log.error("XPENDING 分页失败", ex);
                    return null;
                });
    }

    private List<String> getExpiredIds(List<Object> list) {
        List<String> expiredIds = new ArrayList<>();

        for (Object item : list) {
            List<Object> entry = new ArrayList<>();
            if (item instanceof ArrayList) {
                entry = (ArrayList) item;
            }else if (item instanceof Object[]) {
                Object[] temp = (Object[]) item;
                entry = Arrays.asList(temp);
            }
            String messageId = (String) entry.get(0);
            Long idleTimeMs = (Long) entry.get(2);

            if (idleTimeMs >= 30000) {
                expiredIds.add(messageId);
            }
        }
        return expiredIds;
    }

    private String incrementId(String id) {
        String[] parts = id.split(Pattern.quote("-"));
        long ms = Long.parseLong(parts[0]);
        long seq = Long.parseLong(parts[1]);
        return ms + "-" + (seq + 1);
    }

    private static String lastId(List<Object> list) {
        Object item = list.get(list.size() - 1);
        if (item instanceof ArrayList) {
            ArrayList temp = (ArrayList) item;
            return (String) temp.get(0);
        }else if (item instanceof Object[]) {
            Object[] temp = (Object[]) item;
            return (String) temp[0];
        }
        return "";
    }

    public void shutdown() {
        scheduler.shutdownNow();
        executor.shutdownNow();
    }
}


